package com.guanzi.big_data_mq_process.listener;


import com.alibaba.fastjson.JSON;
import com.guanzi.big_data_mq_process.constant.MessageConstant;
import com.guanzi.big_data_mq_process.model.OrderCreateContext;
import com.guanzi.big_data_mq_process.model.entity.OrderDetails;
import com.guanzi.big_data_mq_process.model.entity.Orders;
import com.guanzi.big_data_mq_process.model.entity.TransactionMessage;
import com.guanzi.big_data_mq_process.model.enums.MessageStatusEnum;
import com.guanzi.big_data_mq_process.service.IOrderDetailsService;
import com.guanzi.big_data_mq_process.service.IOrdersService;
import com.guanzi.big_data_mq_process.service.ITransactionMessageService;
import com.guanzi.big_data_mq_process.service.IUsersService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Service;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.List;

import static com.guanzi.big_data_mq_process.model.enums.LocalTransactionStateEnum.*;
import static org.apache.rocketmq.spring.core.RocketMQLocalTransactionState.*;

/**
 * 事务监听器 属于生产者端
 *  逻辑1：在生产者发送事务消息时，会先发送一个半消息给Broker，Broker接收到并返回给生产者一个成功响应
 *        生产者获取到响应后，会先调用executeLocalTransaction执行本地事务逻辑，根据本地事务状态返回COMMIT、ROLLBACK或者UNKNOWN
 *        如果返回COMMIT，则Broker会提交事务消息，并把半消息删除，然后就跟普通消息一样发送给消费者
 *        返回ROLLBACK，则这条消息相当于不发送，如果返回UNKNOWN，则Broker会反复调用checkLocalTransaction回查事务消息状态，直到方法返回COMMIT或者ROLLBACK
 *  逻辑2：该监听器的逻辑是在生产者端发送给Broker时执行，本身并不会处理消息的内容，还没有执行到消费者。只是我们可以将需要事务包裹的业务可以放在executeLocalTransaction方法执行
 *
 *  具体的消费逻辑在Broker发送给消费者后，消费者去执行
 */
@Slf4j
@Service
@RocketMQTransactionListener
public class OrderTransactionListener implements RocketMQLocalTransactionListener {

    @Autowired
    private IOrdersService ordersService;

    @Autowired
    private IOrderDetailsService orderDetailsService;

    @Autowired
    private IUsersService usersService;

    @Autowired
    private ITransactionMessageService transactionMessageService;

    /**
     * 需要事务包裹的业务放在这里，此处为保存订单信息到数据库里
     *
     * @param message 消息体
     * @param o
     * @return
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        // 事务消息ID，全局唯一，可以用来追踪消息链以及作为第三方支付业务交互的ID
        String messageId = (String) message.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
        log.info("executeLocalTransaction方法被执行，messageId: 【{}】 的header信息为: 【{}】", messageId, message.getHeaders());
        try {
            OrderCreateContext context = (OrderCreateContext) o;
            int orderId = context.getOrderId().intValue();

            // 1、创建主订单
            Orders orders = new Orders();
            orders.setOrderId(orderId);
            orders.setUserId(context.getUserId());
            orders.setOrderDate(new Timestamp(System.currentTimeMillis()));
            orders.setTotalAmount(context.getTotalAmount());
            ordersService.save(orders);

            // 2、创建订单详情
            List<OrderDetails> detailsList = new ArrayList<>();
            for (OrderDetails orderDetails : context.getDetails()) {
                OrderDetails details = new OrderDetails(null, orderId,
                        orderDetails.getProductName(), orderDetails.getQuantity(), orderDetails.getPrice());
                detailsList.add(details);
            }
            orderDetailsService.saveBatch(detailsList);

            // 3、更新用户活跃时间
            usersService.updateUserActiveTime(context.getUserId());

            // 4、保存本地消息表
            TransactionMessage localMsg = new TransactionMessage();
            localMsg.setMessageId(context.getMessageId());
            localMsg.setBizId(String.valueOf(orderId));
            localMsg.setTopic(MessageConstant.ORDER_TOPIC);
            localMsg.setTag(MessageConstant.ORDER_TAG);
            localMsg.setPayload(JSON.toJSONString(message.getPayload()));
            localMsg.setStatus(MessageStatusEnum.WAIT_SENT);
            localMsg.setRetryCount(0);
            localMsg.setNextRetryTime(null);
            localMsg.setCreatedTime(new Timestamp(System.currentTimeMillis()));
            localMsg.setUpdatedTime(new Timestamp(System.currentTimeMillis()));
            localMsg.setTxState(PREPARED);
            transactionMessageService.save(localMsg);
            log.info("executeLocalTransaction被执行，messageId: 【{}】 保存到本地消息表成功!", messageId);

            // 保存到本地消息表和创建完订单之后，返回UNKNOWN，Broker会反复调用checkLocalTransaction回查事务消息状态，直到方法返回COMMIT或者ROLLBACK
            return UNKNOWN;
        } catch (Exception e) {
            return ROLLBACK;
        }
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        String messageId = (String) message.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
        // 回查本地消息表，获取事务状态
        TransactionMessage localMsg = transactionMessageService.getById(messageId);

        log.info("checkLocalTransaction被执行，messageId: 【{}】的当前状态为: {}", messageId, localMsg.getTxState());
        return switch (localMsg.getTxState()) {
            case PREPARED -> UNKNOWN;
            case COMMITTED -> COMMIT;
            case ROLLBACKED -> ROLLBACK;
        };
    }
}
